package cn.acrdpm.config;

import cn.acrdpm.constant.MsgLogStatus;
import cn.acrdpm.service.MsgLogService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;

import javax.annotation.PostConstruct;

/**
 * RabbitConfig:配置类
 *
 * @author 张敏
 * @date 2021/3/1
 */
@Configuration
@PropertySource("classpath:rabbitmq.properties")
@Slf4j
public class RabbitConfig {

    /**
     * 定义交换机及路由键 群发类型的交换机无需定义路由键
     */
    @Value("${com.acrdpm.directexchange}")
    private String directExchange;

    @Value("${com.acrdpm.directroutingkey}")
    private String directRoutingKey;

    @Value("${com.acrdpm.topicexchange}")
    private String topicExchange;

    @Value("${com.acrdpm.topicroutingkey1}")
    private String topicRoutingKey;

    @Value("${com.acrdpm.fanoutexchange}")
    private String fanoutExchange;

    @Value("${com.acrdpm.dieexchange}")
    private String dieExchange;

    @Value("${com.acrdpm.dieroutingkey}")
    private String dieRoutingKey;


    /**
     * 定义直连类型的队列
     */
    @Value("${com.acrdpm.directqueue1}")
    private String directQueue1;

    @Value("${com.acrdpm.directqueue2}")
    private String directQueue2;
    /**
     * 定义主题类型的队列
     */
    @Value("${com.acrdpm.topicqueue1}")
    private String topicQueue1;

    @Value("${com.acrdpm.topicqueue2}")
    private String topicQueue2;

    @Value("${com.acrdpm.topicqueue3}")
    private String topicQueue3;

    @Value("${com.acrdpm.topicqueue4}")
    private String topicQueue4;

    @Value("${com.acrdpm.diequeue}")
    private String dieQueue;

    @Autowired
    RabbitTemplate rabbitTemplate;
    /**
     * 定义群发类型的队列
     */
    @Value("${com.acrdpm.fanoutqueue1}")
    private String fanoutQueue1;

    @Value("${com.acrdpm.fanoutqueue2}")
    private String fanoutQueue2;

    /**
     * 发送邮件
     */
    @Value("${mail_topic_queue}")
    private String mailTopicQueue;

    @Value("${mail_topic_exchange}")
    private String mailTopicExchange;

    @Value("${mail_topic_routingKey}")
    private String mailTopicRoutingKey;

    /**
     * 声明交换机
     */
    @Bean
    public TopicExchange defTopicExchange() {
        return new TopicExchange(topicExchange);
    }

    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange(fanoutExchange);
    }

    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange(directExchange);
    }
    /**
     * 声明队列
     */
     @Bean
     public Queue defTopicQueue1(){
         return  new Queue(topicQueue1,true,false,false);
     }

    /**
     * 声明绑定关系
     */
    @Bean
    public Binding topicQueue1Binding(){
        return BindingBuilder.bind(defTopicQueue1()).to(defTopicExchange()).with(topicRoutingKey);
    }


    @Bean
    public Queue mailQueue() {
        return new Queue(mailTopicQueue, true);
    }

    @Bean
    public TopicExchange mailExchange() {
        return new TopicExchange(mailTopicExchange, true, false);
    }

    @Bean
    public Binding mailBinding() {
        return BindingBuilder.bind( mailQueue()).to(mailExchange()).with(mailTopicRoutingKey);
    }
    /**
     * 消息转化为json格式的配置：
     * 也可在rabbitcontainerConfig 中配置
     */
    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Autowired
    MsgLogService msgLogService;

    /**
     * 定制RabbitTemplate
     * 1、服务收到消息就会回调
     * 1、spring.rabbitmq.publisher-confirms: true
     * 2、设置确认回调
     * 2、消息正确抵达队列就会进行回调
     * 1、spring.rabbitmq.publisher-returns: true
     * spring.rabbitmq.template.mandatory: true
     * 2、设置确认回调ReturnCallback
     * <p>
     * 3、消费端确认(保证每个消息都被正确消费，此时才可以broker删除这个消息)
     * MyRabbitConfig对象创建完成以后，执行这个方法
     */
    @PostConstruct
    public void initRabbitTemplate() {

        /**
         * 1、只要消息抵达Broker就ack=true
         * correlationData：当前消息的唯一关联数据(这个是消息的唯一id)
         * ack：消息是否成功收到
         * cause：失败的原因
         */
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            System.out.println("confirm...correlationData[" + correlationData + "]==>ack:[" + ack + "]==>cause:[" + cause + "]");
            if (ack) {
                log.info("消息成功发送到Exchange");
                String msgId = correlationData.getId();
                msgLogService.updateStatus(msgId, MsgLogStatus.DELIVER_SUCCESS);
            } else {
                log.info("消息发送到Exchange失败, {}, cause: {}", correlationData, cause);
            }
        });

        // 触发setReturnCallback回调必须设置mandatory=true, 否则Exchange没有找到Queue就会丢弃掉消息, 而不会触发回调
        rabbitTemplate.setMandatory(true);
        /**
         * 只要消息没有投递给指定的队列，就触发这个失败回调
         * message：投递失败的消息详细信息
         * replyCode：回复的状态码
         * replyText：回复的文本内容
         * exchange：当时这个消息发给哪个交换机
         * routingKey：当时这个消息用哪个路邮键
         */
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            System.out.println("Fail Message[" + message + "]==>replyCode[" + replyCode + "]" +
                    "==>replyText[" + replyText + "]==>exchange[" + exchange + "]==>routingKey[" + routingKey + "]");

            log.info("消息从Exchange路由到Queue失败: exchange: {}, route: {}, replyCode: {}, replyText: {}, message: {}", exchange, routingKey, replyCode, replyText, message);
        });
    }

}
